What's broken

appendcol pairs the wrong rows together on parallel engines (analytics engine via DataFusion, Spark pushdown), and rows come out in scrambled order.

Internally appendcol lowers to:

main      → ROW_NUMBER() OVER ()
subsearch → ROW_NUMBER() OVER ()
FULL JOIN on (main_rownum = subsearch_rownum)
(no trailing sort)

That only works on a single-threaded engine. On a parallel one, row numbers are assigned arbitrarily and the join then glues the wrong subsearch row onto the wrong main row.

Root cause, step by step (using the IT data)

Step 1 — what appendcol does mechanically

testAppendCol query:

source=account | stats sum(age) as sum by gender, state | sort gender, state
  | appendcol [ stats count(age) as cnt by gender | sort gender ]
  | fields gender, state, sum, cnt | head 10

Main (after stats sum(age) by gender, state | sort gender, state) — many rows, one per (gender, state). F rows come first alphabetically, then M rows:

F/AK   sum=317
F/AL   sum=397
F/AR   sum=229
F/AZ   sum=238
…
M/AK   sum=…
M/MD   sum=580
…

Subsearch (stats count(age) by gender | sort gender) — exactly 2 rows:

F   cnt=493     ← count of all F docs
M   cnt=507     ← count of all M docs

appendcol (1) numbers each main row, (2) numbers each subsearch row, (3) FULL JOINs on those row numbers.

Step 2 — on v2 (serial), row numbers follow input order

Main:    _rn_main_       Sub:        _rn_sub_
F/AK     1               F  cnt=493  1
F/AL     2               M  cnt=507  2
F/AR     3
…

Join on _rn_main_ = _rn_sub_:

After head 10: F/AK, F/AL, F/AR, F/AZ, F/CA, F/CO, F/CT, F/DC, F/DE, F/FL — exactly what the IT asserts.

Step 3 — on the analytics route (DataFusion, parallel), row numbers are arbitrary

ROW_NUMBER() OVER () ignores input order on a parallel engine. Example random assignment:

Main:    _rn_main_         Sub:        _rn_sub_
F/AK     13   ← arbitrary  F  cnt=493  2   ← arbitrary
F/AL     5                 M  cnt=507  1
F/AR     28
…
M/MD     1    ← happened to get 1

Same FULL JOIN:

This matches the actual IT failure:

Expected: [F,AK,317,493], [F,AL,397,507], [F,AR,229,null], …, [F,FL,310,null]
Actual contained: ["M","MD",580,507]    ← M row leaked into top-10

M/MD leaked into the top 10 (sort order was lost too), and the cnt values landed on the wrong main rows entirely.

Step 4 — a downstream sort can't fix this

What if we add | sort gender, state at the end to "fix the order"?

F/AK, sum=317, cnt=null     ← still wrong, IT wants 493
F/AL, sum=397, cnt=null     ← still wrong, IT wants 507
F/AR, sum=229, cnt=null
…
M/MD, sum=580, cnt=507      ← the wrong-attached cnt is still attached

A sort only reorders tuples; it can't pry one open and swap cnt values between rows. The exact tuple [F, AK, 317, 493] the IT asserts doesn't exist anywhere in the broken output — no row anywhere has F/AK paired with 493. No verifyDataRowsInAnyOrder or trailing sort can recover that.

Step 5 — the fix, applied to the same scenario

The wrong pairing is committed inside visitAppendCol (Step 3). Once those tuples exist, no downstream PPL command can undo them — so the lowering itself must produce deterministic row numbers. Two changes do that:

(a) Fill the window's OVER (???) with the upstream sort's collation:

Main:      ROW_NUMBER() OVER (ORDER BY gender, state)
Subsearch: ROW_NUMBER() OVER (ORDER BY gender)

DataFusion is now obliged to sort by those keys before numbering, so it assigns the same numbers v2 would:

Main:    _rn_main_       Sub:        _rn_sub_
F/AK     1               F  cnt=493  1
F/AL     2               M  cnt=507  2
F/AR     3
F/AZ     4
…

The FULL JOIN on row numbers now pairs correctly: F/AK ↔ F(493), F/AL ↔ M(507), rest null. The tuples are right — [F, AK, 317, 493] and [F, AL, 397, 507] exist in memory.

(b) Add a trailing sort by _rn_main_, _rn_sub_ (NULLS LAST) after the join:

DataFusion's hash join can still scramble output order even when the individual tuples are correct. The trailing sort puts them back in row-number order, which (by Piece a) is the upstream sort order:

F/AK, sum=317, cnt=493     ← _rn_main_ = 1
F/AL, sum=397, cnt=507     ← _rn_main_ = 2
F/AR, sum=229, cnt=null    ← _rn_main_ = 3
F/AZ, sum=238, cnt=null    ← _rn_main_ = 4
…

After head 10: the same 10 F-rows v2 produced (F/AK … F/FL), with the right cnt on each. The IT now passes 2/2 on the analytics route.

Both pieces are required — (a) alone leaves the post-join output scrambled by the hash join; (b) alone sorts by row numbers that are themselves random.

Where the ROW_NUMBER() OVER (...) is emitted

The window expression is built inside visitAppendCol via a helper PlanUtils.makeOver:

RexNode mainRowNumber = PlanUtils.makeOver(
    context,
    BuiltinFunctionName.ROW_NUMBER,
    null,
    List.of(),
    List.of(),                       // partition keys (empty)
    mainOrderKeys,                   // ← the ORDER BY list — this is what we change
    WindowFrame.toCurrentRow());

context.relBuilder.projectPlus(
    context.relBuilder.alias(mainRowNumber, ROW_NUMBER_COLUMN_FOR_MAIN));

End-to-end path of the offending construct:

PPL "| appendcol [...]"
   ↓  parser → AST AppendCol
visitAppendCol → PlanUtils.makeOver(... orderKeys=??? ...)
   ↓  produces a RexOver { ROW_NUMBER, OVER(orderKeys=???) }
LogicalProject(..., _row_number_main_=[ROW_NUMBER() OVER (???)])
   ↓  Calcite → Substrait
DataFusion's window operator runs it

Before the fix, mainOrderKeys was hard-coded to List.of(), so DataFusion saw bare OVER () and assigned numbers arbitrarily.

Fix

Two changes in visitAppendCol:

1. Fill the ???: change mainOrderKeys from hard-coded List.of() to deriveCollationOrderKeys(context), which reads the upstream collation via RelMetadataQuery.collations(peek()). So OVER () becomes OVER (ORDER BY <upstream sort>) — deterministic on DataFusion.
2. Trailing sort by the row-number columns after the FULL JOIN, so the final output order doesn't depend on join execution (same pattern streamstats already uses).

No behavior change on v2/Calcite (its serial execution already produced this order).

Results

CalcitePPLAppendcolIT:

Testing

• CalcitePPLAppendcolTest (5 unit tests, regenerated plans) ✅
• CalcitePPLAppendcolIT 2/2 on both paths ✅
• NewAddedCommandsIT.testAppendcol ✅
• appendcol.md doctest ✅ (no doc edits needed)
• spotlessCheck clean ✅

Check List

• New functionality includes testing.
• New functionality has been documented (n/a — behavior-preserving fix).
• Commits are signed per the DCO using --signoff.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.